30. K-Means算法-股市波动聚类

from pathlib import Path  # 导入路径工具以探测课程数据挂载
import numpy as np  # 导入数值计算库
import pandas as pd  # 导入数据分析库
from sklearn.cluster import KMeans  # 导入K-Means聚类模型
from sklearn.metrics import silhouette_score  # 导入轮廓系数评估指标
candidate_roots = [Path('/home/ubuntu/r2_data_mount/data'), Path(r'C:\qiufei\data')]  # 按课程规定顺序探测数据根目录
data_root = next((root for root in candidate_roots if root.is_dir()), None)  # 选择第一个可用根目录
index_path = data_root / 'index' / 'hs300_index_only.h5' if data_root else None  # 定位沪深300本地行情文件
if index_path is None or not index_path.is_file():  # 真实数据缺失时明确提示
    raise FileNotFoundError(f'未找到沪深300数据文件：{index_path}；请从课程数据下载入口获取')  # 提示读者下载文件
hs300_daily = pd.read_hdf(index_path, key='hs300')  # 读取真实指数日行情
trade_date = pd.to_datetime(hs300_daily['datetime'].astype(str).str[:8], format='%Y%m%d')  # 解析八位交易日字符串
daily_log_return = np.log(hs300_daily['close'] / hs300_daily['prev_close'])  # 用收盘与昨收计算对数收益
daily_log_return.index = trade_date  # 以交易日作为收益序列索引
window_sample = daily_log_return.sort_index().loc['2013-01-01':'2026-06-30'].dropna()  # 截取课堂窗口样本期
half_year_code = window_sample.index.year * 10 + (window_sample.index.month > 6).astype(int) + 1  # 为每个交易日标记所属半年
window_stats = window_sample.groupby(half_year_code).agg(['count', 'mean', 'std'])  # 聚合每个半年的交易日数/均值/标准差
minimum_window_days = 60  # 预声明的完整半年窗口最少交易日
window_stats = window_stats[window_stats['count'] >= minimum_window_days]  # 过滤残缺窗口（如仅约20个交易日的2026H1）
window_features = pd.DataFrame({'annualized_return': window_stats['mean'] * 252, 'annualized_volatility': window_stats['std'] * np.sqrt(252)})  # 构造年化特征
def format_half_year(code):  # 将半年编码转为可读标签
    return f'{code // 10}H{code % 10}'  # 前四位为年份，末位1或2为半序
print(f'输入口径：{len(window_features)}个完整半年窗口（{format_half_year(window_features.index[0])}—{format_half_year(window_features.index[-1])}，各≥{minimum_window_days}个交易日），两特征')  # 报告样本边界

from sklearn.preprocessing import StandardScaler  # 导入可反变换的标准化器
split_position = int(len(window_features) * 0.70)  # 按时间顺序固定前70%训练、后30%留出
train_features = window_features.iloc[:split_position].copy()  # 仅用早期窗口拟合模型
holdout_features = window_features.iloc[split_position:].copy()  # 隔离后期留出窗口
feature_scaler = StandardScaler().fit(train_features)  # 只在训练期拟合均值与尺度
train_z = pd.DataFrame(feature_scaler.transform(train_features), index=train_features.index, columns=train_features.columns)  # 转换训练特征
silhouette_by_k = {}  # 存放每个K的轮廓系数
for cluster_count in range(2, 6):  # 扫描K=2到5
    fitted_labels = KMeans(n_clusters=cluster_count, n_init=10, random_state=42).fit_predict(train_z)  # 只在训练期固定随机种子拟合
    silhouette_by_k[cluster_count] = silhouette_score(train_z, fitted_labels)  # 记录训练期轮廓系数
best_k = max(silhouette_by_k, key=silhouette_by_k.get)  # 取轮廓系数最高的K
kmeans_model = KMeans(n_clusters=best_k, n_init=10, random_state=42).fit(train_z)  # 用训练期最优K拟合固定质心
train_labels = pd.Series(kmeans_model.labels_, index=train_z.index)  # 保存训练标签
train_distance = kmeans_model.transform(train_z).min(axis=1)  # 计算训练样本到所属质心的距离
distance_reference = float(np.quantile(train_distance, 0.95))  # 以训练距离95%分位数作为固定参照
cluster_share = train_labels.value_counts(normalize=True)  # 训练期各簇占比
silhouette_limit, share_limit = 0.30, 0.15  # 预声明训练分离度与最小簇课堂筛查标准
train_conditions_met = silhouette_by_k[best_k] >= silhouette_limit and cluster_share.min() >= share_limit  # 合并两项训练结果判断
print('各K轮廓系数：' + '，'.join(f'K={k}:{v:.3f}' for k, v in silhouette_by_k.items()))  # 展示K扫描结果
print(f'训练截止{format_half_year(train_features.index[-1])}｜留出起点{format_half_year(holdout_features.index[0])}｜最优K={best_k}｜训练轮廓{silhouette_by_k[best_k]:.3f}')  # 展示时间边界与训练分离
print(f'训练最小簇{cluster_share.min():.1%}｜训练结果判断={train_conditions_met}')  # 展示训练期检查要求依据

holdout_z = pd.DataFrame(feature_scaler.transform(holdout_features), index=holdout_features.index, columns=holdout_features.columns)  # 用训练期固定参数转换留出特征
holdout_labels = pd.Series(kmeans_model.predict(holdout_z), index=holdout_z.index)  # 把留出窗口分配到固定质心
cluster_labels = pd.concat([train_labels, holdout_labels]).sort_index()  # 合并绘图标签但不重新拟合
holdout_distance = kmeans_model.transform(holdout_z).min(axis=1)  # 计算留出样本到所属质心的距离
holdout_within_reference = float((holdout_distance <= distance_reference).mean())  # 计算留出距离覆盖率
feature_drift = holdout_z.mean().sub(train_z.mean()).abs().to_dict()  # 报告标准化特征均值漂移
coverage_limit = 0.80  # 预声明固定质心距离覆盖课堂筛查标准
holdout_conditions_met = holdout_labels.nunique() >= 2 and holdout_within_reference >= coverage_limit  # 合并留出覆盖与状态数门
conditions_met = train_conditions_met and holdout_conditions_met  # 训练或留出任一依据失败即关闭描述状态
decision = '仅作课堂描述' if conditions_met else '暂不采用模型'  # 只给出描述性结果；数据要求未满足时不继续计算
print(f'训练最小簇{cluster_share.min():.1%}｜留出距离覆盖{holdout_within_reference:.1%}｜特征漂移{feature_drift}')  # 展示留出分配距离与漂移
print(f'检查结果：{decision}｜留出覆盖阈值敏感性需比较70%/80%/90%')  # 展示判断结果与敏感性要求